package com.lagou.hbase.client;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.exceptions.DeserializationException;
import org.apache.hadoop.hbase.security.visibility.CellVisibility;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.NavigableMap;
import java.util.Set;

public class HbaseClientDemo {
    Configuration conf;
    Connection conn;

    @Before
    public void init() throws IOException {
        //获取配置文件
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "linux121,linux122");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        conn = ConnectionFactory.createConnection(conf);
    }

    //创建表
    @Test
    public void createTable() throws IOException {
        HBaseAdmin admin = (HBaseAdmin) conn.getAdmin();
        HTableDescriptor worker = new HTableDescriptor(TableName.valueOf("relationship"));
        //指定列族
        worker.addFamily(new HColumnDescriptor("friends"));
        admin.createTable(worker);
        System.out.println("worker表创建成功");
    }

    //创建表
    @Test
    public void deleteTable() throws IOException {
        HBaseAdmin admin = (HBaseAdmin) conn.getAdmin();
        admin.disableTable(TableName.valueOf("relationship"));
        admin.deleteTable(TableName.valueOf("relationship"));
        System.out.println("relationship成功");
    }

    //插入一条数据
    @Test
    public void putData() throws IOException {
        Table worker = conn.getTable(TableName.valueOf("worker"));
        Put put = new Put(Bytes.toBytes("110"));//指定rowkey
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("addr"),Bytes.toBytes("beijing"));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"),Bytes.toBytes("laowang"));
        worker.put(put);
        worker.close();
        System.out.println("插入数据到worker表成功");
    }

    @Test
    public void deleteData() throws IOException, DeserializationException {
        Table worker = conn.getTable(TableName.valueOf("relationship"));
        Delete delete = new Delete(Bytes.toBytes("uid1"));
        delete.deleteColumns(Bytes.toBytes("friends"), Bytes.toBytes("uid4"));
       // delete.addColumn(Bytes.toBytes("info"), Bytes.toBytes("addr"));
        byte[] row = delete.getRow();
        System.out.println(Bytes.toString(row));
        Set<String> strings = delete.getAttributesMap().keySet();
        NavigableMap<byte[], List<Cell>> familyCellMap = delete.getFamilyCellMap();
        for (byte[] bytes : familyCellMap.keySet()) {
            System.out.println(Bytes.toString(bytes));
            System.out.println(familyCellMap.get(bytes));
            List<Cell> cells = familyCellMap.get(bytes);
            for (Cell cell : cells) {
                System.out.println("family:"+Bytes.toString(CellUtil.cloneFamily(cell)));
                System.out.println("value:"+Bytes.toString(CellUtil.cloneValue(cell)));
                System.out.println("row:"+Bytes.toString(CellUtil.cloneRow(cell)));
                System.out.println("qualifier:"+Bytes.toString(CellUtil.cloneQualifier(cell)));

            }
        }
        CellVisibility cellVisibility = delete.getCellVisibility();
        for (byte[] bytes : familyCellMap.keySet()) {
            System.out.println(Bytes.toString(bytes));
        }
        System.out.println(strings);
        worker.delete(delete);
        worker.close();
        System.out.println("worker表删除数据成功");
    }

    @Test
    public void deletes1() throws IOException {
        FileOutputStream test = new FileOutputStream("ceshi3");
        test.write("test3".getBytes(StandardCharsets.UTF_8));
        test.write("\r\n".getBytes(StandardCharsets.UTF_8));
        test.flush();
        HTableInterface htable = (HTableInterface) conn.getTable(TableName.valueOf("relationship"));
        Delete delete = new Delete(Bytes.toBytes("uid1"));
        delete.deleteColumns(Bytes.toBytes("friends"), Bytes.toBytes("uid4"));

        byte[] row = delete.getRow();
        test.write("htableinterface".getBytes(StandardCharsets.UTF_8));
        test.write("\r\n".getBytes(StandardCharsets.UTF_8));
        NavigableMap<byte[], List<Cell>> familyCellMap = delete.getFamilyCellMap();
        test.write((familyCellMap.keySet().size() + "").getBytes(StandardCharsets.UTF_8));
        test.write("\r\n".getBytes(StandardCharsets.UTF_8));
        int i = 0;
        List<Delete> list = new ArrayList<>();
        for (byte[] bytes : familyCellMap.keySet()) {            // bytes列族
            List<Cell> cells = familyCellMap.get(bytes);
            test.write("familyCellMap.get(bytes)".getBytes(StandardCharsets.UTF_8));
            test.write("\r\n".getBytes(StandardCharsets.UTF_8));
            for (Cell cell : cells) {
                i++;
                test.write(bytes);
                test.write("\r\n".getBytes());
                test.write(CellUtil.cloneQualifier(cell));
                test.write("\r\n".getBytes());
                test.flush();
                test.write(bytes);
                test.write("\t ceshi".getBytes(StandardCharsets.UTF_8));
                test.write(row);
                Get get = new Get(CellUtil.cloneQualifier(cell));
                //指定查询某个列族
                get.addFamily(Bytes.toBytes("friends"));
                Result result = htable.get(get);
                Cell[] rawCells = result.rawCells();
                for (Cell rawCell : rawCells) {
                    if (Bytes.toString(CellUtil.cloneQualifier(rawCell)).equals(Bytes.toString(delete.getRow()))) {
                        //是需要删除的数据
                        Delete delete1 = new Delete(CellUtil.cloneQualifier(cell));
                        delete1.addColumns(bytes, row);
                        list.add(delete1);

                    }

                }
                test.write("new delete完成".getBytes(StandardCharsets.UTF_8));
                test.write("\r\n".getBytes(StandardCharsets.UTF_8));
                test.write("delete.deletecolumn完成".getBytes(StandardCharsets.UTF_8));
                test.write("\r\n".getBytes(StandardCharsets.UTF_8));
            }
        }
        htable.delete(list);
        test.write(("i===>" + i).getBytes(StandardCharsets.UTF_8));
        test.write("\r\n".getBytes(StandardCharsets.UTF_8));
        htable.close();
        test.close();
    }

    //查询数据
    @Test
    public void getData() throws IOException {
        Table worker = conn.getTable(TableName.valueOf("relationship"));
        Get get = new Get(Bytes.toBytes("uid1"));
        //指定查询某个列族
        get.addFamily(Bytes.toBytes("friends"));
        Result result = worker.get(get);
        Cell[] cells = result.rawCells();
        for (Cell cell : cells) {
            String row = Bytes.toString(CellUtil.cloneRow(cell));
            String family = Bytes.toString(CellUtil.cloneFamily(cell));
            String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
            String value = Bytes.toString(CellUtil.cloneValue(cell));
            System.out.println("row:"+row);//row key
            System.out.println("family:"+family);//列族
            System.out.println("qualifier:"+qualifier);//列名
            System.out.println("value:"+value);//列值

        }

    }

    //全表扫描
    @Test
    public void scanData() throws IOException {
        Table worker = conn.getTable(TableName.valueOf("worker"));
        Scan scan = new Scan();
        ResultScanner scanner = worker.getScanner(scan);
        for (Result result : scanner) {
            Cell[] cells = result.rawCells();
            for (Cell cell : cells) {
                String row = Bytes.toString(CellUtil.cloneRow(cell));
                String family = Bytes.toString(CellUtil.cloneFamily(cell));
                String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.println("row:"+row);
                System.out.println("family:"+family);
                System.out.println("qualifier:"+qualifier);
                System.out.println("value:"+value);

            }
        }

    }

    //区间范围扫描
    @Test
    public void scanRange() throws IOException {
        Table worker = conn.getTable(TableName.valueOf("worker"));
        Scan scan = new Scan(Bytes.toBytes("001"), Bytes.toBytes("004"));
        ResultScanner scanner = worker.getScanner(scan);
        for (Result result : scanner) {
            Cell[] cells = result.rawCells();
            for (Cell cell : cells) {
                String row = Bytes.toString(CellUtil.cloneRow(cell));
                String family = Bytes.toString(CellUtil.cloneFamily(cell));
                String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.println("row:"+row);
                System.out.println("family:"+family);
                System.out.println("qualifier:"+qualifier);
                System.out.println("value:"+value);

            }
        }
    }


    @After//释放连接
    public void release() throws IOException {
        if (conn != null) {
            conn.close();
        }
    }
}
